package cn.doitedu.rtdw.dash_board;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job2_流量看板指标2_分组topn {
    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        //  创建kafka中的维度打宽的行为日志的明细数据，逻辑映射表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                          "
                        +"     user_id           BIGINT,                     "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_Id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "
                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "
                        +"     url_prefix    STRING,                         "
                        +"     page_type    STRING,                          "
                        +"     page_service STRING,                          "

                        +"     proc_time AS proctime(),                      " // processing time 时间语义
                        +"     rt AS  to_timestamp_ltz(event_time,3),        " // 表达式字段，用于将event_time转成timestamp(3)类型
                        +"     WATERMARK FOR rt AS  rt - INTERVAL '0' SECOND   " // 基于rt字段定义watermark，从此，rt字段就具备了 flink中事件时间语义

                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd_events',                          "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'latest-offset',           "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           "
        );


        // 建表，映射mysql中的目标表
        tenv.executeSql(
                " create table dash_uv_topn_page_5m_mysql(         "+
                        "   page_type    string                            "+
                        "   ,`rank`         bigint                           "+
                        "   ,window_start timestamp                        "+
                        "   ,window_end   timestamp                        "+
                        "   ,page_url     string                           "+
                        "   ,uv           bigint                           "+
                        "   ,primary key(page_type,`rank`) not enforced      "+
                        " ) with (                                         "+
                        "     'connector' = 'jdbc',                        "+
                        "     'url' = 'jdbc:mysql://doitedu:3306/doit41',  "+
                        "     'table-name' = 'dash_uv_topn_page_5m',       "+
                        "     'username' = 'root',                         "+
                        "     'password' = 'root'                          "+
                        " )                                                "
        );


        // 统计sql
        tenv.executeSql(
                "INSERT INTO dash_uv_topn_page_5m_mysql  " +
                        "SELECT                                                                                          "+
                        "     page_type,                                                                                  "+
                        "     rn as `rank`,                                                                               "+
                        "     window_start,                                                                               "+
                        "     window_end,                                                                                 "+
                        "     page_url,                                                                                   "+
                        "     uv                                                                                          "+
                        " FROM (                                                                                          "+
                        "     SELECT                                                                                      "+
                        "         window_start,                                                                           "+
                        "         window_end,                                                                             "+
                        "         page_type,                                                                              "+
                        "         page_url,                                                                               "+
                        "     	uv,                                                                                       "+
                        "     	row_number() over(partition by window_start,window_end,page_type order by uv desc) as rn  "+
                        "     FROM (                                                                                      "+
                        "         SELECT                                                                                  "+
                        "             window_start,                                                                       "+
                        "         	window_end,                                                                           "+
                        "         	page_type,                                                                            "+
                        "         	properties['url'] as page_url,                                                        "+
                        "         	count(distinct user_id) as uv                                                         "+
                        "         FROM TABLE(                                                                             "+
                        "             TUMBLE(TABLE dwd_kafka ,DESCRIPTOR(rt),INTERVAL '5' MINUTE)                         "+
                        "         )                                                                                       "+
                        "         GROUP BY                                                                                "+
                        "             window_start,                                                                       "+
                        "         	window_end,                                                                           "+
                        "         	page_type,                                                                            "+
                        "         	properties['url']                                                                     "+
                        "     ) o1                                                                                        "+
                        " ) o2                                                                                            "+
                        " WHERE rn<=2                                                                                     "
        );
    }

}
